
//##include "nosal/exec/internal/execution_manager.h"
//##include "nosal/exec/internal/serialization.h"
//##include "nosal/exec/internal/execution_ordering.h"
//##include "nosal/exec/internal/exception.h"
//##include "nosal/exec/internal/filesystem.h"
//##include "nosal/exec/internal/constant.h"
//##include "nosal/exec/internal/logger.h"
//##include "nosal/exec/internal/local_socket.h"

#include "common/constant.h"
#include "common/exception.h"
#include "execution_manager.h"
#include "executables_parser.h"
#include "executables_handler.h"
#include "logger.h"
#include "nosal/log/logger.h"
#include "nosal/log/logging.h"
//#include "nosal/exec/state_client.h"
//#include "nosal/exec/execution_client.h"

//#include "apd/manifestreader/manifest_reader.h"

#include <set>
#include <list>
#include <sstream>
#include <iostream>
#include <thread>
#include <chrono>

#include <cstdio>  // std::remove
#include <dirent.h>
#include <poll.h>
#include <signal.h>
//#include <sys/reboot.h>
//#include <libcgroup.h>
#include <sys/mount.h>
#include <sys/stat.h>
//#include "nosal/core/string.h"
#include <string>
#include <cstring>

namespace netaos {
namespace nosal {
namespace exec {
namespace internal {

namespace
{
const char* const kMachineManifestPath = "/etc/system/machine_manifest.json";

netaos::nosal::log::Logger& GetLogger()
{
    static netaos::nosal::log::Logger& logger = netaos::nosal::log::CreateLogger("EM", "ExecutionManager", netaos::nosal::exec::internal::kLogLevel);
    return logger;
}

}  // namespace

ExecutionManager::ExecutionManager(const char* root_path)
    : root_path_(root_path)
    , platform_state_(PlatformStates::kRunning)
    , state_change_handler_(function_groups_, executed_startup_configurations_)
{
    GetLogger().LogInfo() << "Initializing Execution Manager.";
    Initialize();
}

ExecutionManager::~ExecutionManager()
{
    GetLogger().LogDebug() << "Syncing...";
    //::sync();

    // Free the lookup before closing
    //findProcess_.reset();

    if (platform_state_ == PlatformStates::kShuttingdown && root_path_ == "/") {

        GetLogger().LogInfo() << "Requesting RB_POWER_OFF from the kernel";
      //  reboot(RB_POWER_OFF);
    } else if (platform_state_ == PlatformStates::kRestarting && root_path_ == "/") {

        GetLogger().LogInfo() << "Requesting RB_AUTOBOOT from the kernel";
       // reboot(RB_AUTOBOOT);
    }
    // The following code can be reached only if the system Shutdown or system Restart
    // are not configured properly (for example when you don't want it).
    GetLogger().LogInfo() << "Execution Manager exits.";
}

int ExecutionManager::Run(const char* root_path) noexcept
{
    try {
        GetLogger().LogInfo() << "Execution Manager uses root_path" << root_path;

        ExecutionManager instance(root_path);

        GetLogger().LogInfo() << "Execution Manager is running.";
        instance.Run();

        return EXIT_SUCCESS;
    } catch (const std::exception& ex) {
        GetLogger().LogFatal() << "Caught exception:" << ex.what();
    } catch (...) {
        GetLogger().LogFatal() << "Caught unknown exception";
    }

    GetLogger().LogInfo().Flush();

    return EXIT_FAILURE;
}
void ExecutionManager::TerminateAllProcesses(bool kill)
{
    for (auto executed_startup_config : executed_startup_configurations_) {
        if (kill) {
            executed_startup_config->Kill();
        } else {
            executed_startup_config->Term();
        }
    }

    std::chrono::nanoseconds wait_for_termination = manifest_->GetDefaultProcessTimeout().GetExitTimeout();
    std::this_thread::sleep_for(wait_for_termination);

    //StartupConfiguration::CheckProcesses(executed_startup_configurations_);

    auto it = executed_startup_configurations_.begin();
    while (it != executed_startup_configurations_.end()) {
        if ((*it)->GetState() == ProcessState::kTerminated) {
            it = executed_startup_configurations_.erase(it);
        } else {
            it++;
        }
    }
}

void ExecutionManager::EnterUnrecoverableState()
{
    // pre-cleanup:
    // clear all processed SetState Requests and set flag for ongoing state transitions to false.
    GetLogger().LogFatal() << "Execution Manager enters Unrecoverable State.";
    //processed_requests_.clear();
    ongoing_state_transition_ = false;

    // try to terminate and then kill all processes until all are gone.
    TerminateAllProcesses(false);
    while (!executed_startup_configurations_.empty()) {
        TerminateAllProcesses(true);
    }

    // post cleanup:
    // set PlatformState to kUnrecoverableState.
    platform_state_ = PlatformStates::kUnrecoverableState;
    std::cerr << "Execution Manager is in Unrecoverable state." << std::endl;
}

void ExecutionManager::Run()
{
    GetLogger().LogInfo() << "Performing initial switch to Machine State Startup.";

    
    ProcessedSetStateRequest startupRequest
        = ProcessedSetStateRequest(kMachineFunctionGroupName, config::kStateMachineStartup);
    //StateClientImpl::SetStateResponse::Status startupResponseStatus
    auto result   = state_change_handler_.SetCompleteState(startupRequest);

    //if ((startupResponseStatus == StateClientImpl::SetStateResponse::Status::kSuccess)
    if (result
        && (function_groups_.at(kMachineFunctionGroupName).GetCurrentStateName() == "Startup")) {
        was_initial_machine_state_transition_successful_ = true;
    } else {
        was_initial_machine_state_transition_successful_ = false;
        EnterUnrecoverableState();
    }

    while (PlatformStates::kRunning == platform_state_) {
#if 0
        // Receive Execution State reports (kRunning) and update related processes.
        ProcessExecutionStateReports();

        // Refresh FGs states in case of unexpected termination of a process
        UpdateFunctionGroupsStates();

        // Receive requests from StateClient and execute them.
        //     - the requests are processed synchronously because we need to return their results;
        //     - process errors are detected, reported and handled in CheckProcesses().
        ProcessStateClientRequests();

        // Process any ongoing State Transitions
        ProcessStateTransitions();
#endif
        if (executed_startup_configurations_.empty()) {
            GetLogger().LogWarn() << "There are no more child processes.";
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(2));
    }

    if (platform_state_ == PlatformStates::kUnrecoverableState) {
        while (true) {
            std::cerr << "Execution Manager is in Unrecoverable State." << std::endl;
            std::this_thread::sleep_for(std::chrono::seconds(30));
        }
    }

    GetLogger().LogInfo() << "Execution Manager is stopping.";
}
#if 0
std::shared_ptr<StartupConfiguration> ExecutionManager::FindStartupConfiguration(pid_t pid) const
{
    auto hasPid = [pid](const std::shared_ptr<StartupConfiguration> process) { return process->GetPid() == pid; };

    const auto it
        = std::find_if(executed_startup_configurations_.cbegin(), executed_startup_configurations_.cend(), hasPid);

    if (it != executed_startup_configurations_.cend()) {
        return *it;
    }

    GetLogger().LogWarn() << "Failed to find a process with id" << pid;
    return nullptr;
}
#endif
void ExecutionManager::Initialize()
{
    GetLogger().LogInfo() << "Start of Execution Manager initialization.";

    LoadMachineConfiguration();

   // InitializeCgroups();

    LoadExecutionManifests();

    //InitializeFindProcess();

    GetLogger().LogInfo() << "Initialization finished";
}
#if 0
void ExecutionManager::RejectStateClientRequest(const PipeMessage& requestMessage,
    std::shared_ptr<StartupConfiguration> requestor)
{
    GetLogger().LogError() << "Rejecting a StateClient request from process" << requestor->GetName()
                           << "which is not a StateManagement process.";
    switch (requestMessage.header.tag) {
    case StateClientOperation::kSetStateRequest: {
        StateClientImpl::SetStateResponse response{
            StateClientImpl::SetStateResponse::Status::kNotInvokedByStateManagement, {}};
        StateClientImpl::SetStateRequest request;
        UnpackSetStateRequest({requestMessage.data.data(), requestMessage.data.size()}, request);
        SendSetStateResponse(requestor, response, request.responseTimeout);
        break;
    }

    case StateClientOperation::kGetInitialMachineStateTransitionRequest: {
        StateClientImpl::GetInitialMachineStateTransitionResponse response{
            StateClientImpl::GetInitialMachineStateTransitionResponse::Status::kNotInvokedByStateManagement};
        SendGetInitialMachineStateTransitionResponse(requestor, response);
        break;
    }

    case StateClientOperation::kGetExecutionErrorRequest: {
        StateClientImpl::GetExecutionErrorResponse response{
            StateClientImpl::GetExecutionErrorResponse::Status::kNotInvokedByStateManagement, {}};
        SendGetExecutionErrorResponse(requestor, response);
        break;
    }

    default: {
        SendUnknownResponse(requestor);
        break;
    }
    }
}
#endif

void ExecutionManager::ProcessStateTransitions()
{
    if (!ongoing_state_transition_) {
        return;
    }

    auto it = processed_requests_.begin();
    while (it != processed_requests_.end()) {
        bool finished = ProcessSetStateRequest(*it);

        if (finished) {
            GetLogger().LogDebug() << "Finished with SetStateRequest from StateClient" << it->requestor->GetName() + "."
                                   << "Removing from List of processed Requests.";
            it = processed_requests_.erase(it);
        } else {
            ++it;
        }
    }

    if (processed_requests_.size() == 0) {
        ongoing_state_transition_ = false;
    }
}

#if 0
bool ExecutionManager::CheckIfAlreadyInState(const StateClientImpl::StateChange& stateChange)
{
    const auto it = function_groups_.find(stateChange.functionGroupName);

    return (it != function_groups_.end()) && (it->second.GetCurrentStateName() == stateChange.newStateName);
}
void ExecutionManager::ProcessStateClientRequests()
{
    const auto& pollResult = StartupConfiguration::PollForStateClientRequests(executed_startup_configurations_);
    for (const auto& entry : pollResult) {

        const PipeMessage& requestMessage = entry.first;
        std::shared_ptr<StartupConfiguration> requestor = entry.second;
        const std::string packedRequest(requestMessage.data.data(), requestMessage.data.size());

        GetLogger().LogInfo() << "Handling a message from StateClient" << requestor->GetName();

        if (requestor->GetFunctionClusterAffiliation() != kStateManagementFunctionClusterAffiliation) {
            RejectStateClientRequest(requestMessage, requestor);
        }

        switch (requestMessage.header.tag) {
        case StateClientOperation::kSetStateRequest: {

            GetLogger().LogInfo() << "Received a SetState request from" << requestor->GetName() << "with pid"
                                  << requestor->GetPid();

            StateClientImpl::SetStateRequest request;
            UnpackSetStateRequest(packedRequest, request);

            if (ongoing_state_transition_) {
                auto it = std::find_if(processed_requests_.cbegin(),
                    processed_requests_.cend(),
                    [&request](const ProcessedSetStateRequest& processedRequest) {
                        return request.change.functionGroupName == processedRequest.request.change.functionGroupName;
                    });

                if (it != processed_requests_.end()) {
                    if (it->request.change.newStateName == request.change.newStateName) {
                        GetLogger().LogInfo()
                            << "Rejecting SetState request of FunctionGroup" << request.change.functionGroupName
                            << "to state" << request.change.newStateName
                            << "because the transition is already ongoing.";
                        /// @uptrace{SWS_EM_02554, 825642f94154ef3bd909ad8f7ff56934e2d08796}
                        StateClientImpl::SetStateResponse response{
                            StateClientImpl::SetStateResponse::kAlreadyInTransitionToSameState, {}};
                        SendSetStateResponse(requestor, response, request.responseTimeout);
                        break;
                    } else {
                        GetLogger().LogInfo()
                            << "Cancelling transition of FunctionGroup" << (*it).request.change.functionGroupName
                            << "to state" << (*it).request.change.newStateName << "because of newer request to state"
                            << request.change.newStateName;
                        /// @uptrace{SWS_EM_02298, bd990b9881c06419c1fb9e30ae522112268893eb}
                        StateClientImpl::SetStateResponse response{
                            StateClientImpl::SetStateResponse::Status::kCancelled, {}};
                        SendSetStateResponse((*it).requestor, response, request.responseTimeout);
                        processed_requests_.erase(it);
                    }
                }
            }

            if (CheckIfAlreadyInState(request.change)) {
                GetLogger().LogInfo() << "Rejecting SetState request of FunctionGroup"
                                      << request.change.functionGroupName << "to state" << request.change.newStateName
                                      << "because the FunctionGroup is already in this state.";
                /// @uptrace{SWS_EM_02553, 705a0e3e342a68d4da64deb60e35c9ebf0dfee7c}
                StateClientImpl::SetStateResponse response{
                    StateClientImpl::SetStateResponse::Status::kAlreadyInState, {}};
                SendSetStateResponse(requestor, response, request.responseTimeout);
                break;
            }

            ProcessedSetStateRequest processedRequest{};
            processedRequest.request = request;
            processedRequest.requestor = requestor;
            processed_requests_.emplace_back(processedRequest);
            ongoing_state_transition_ = true;

            break;
        }

        case StateClientOperation::kGetExecutionErrorRequest: {

            GetLogger().LogInfo() << "Received a GetExecutionError request from" << requestor->GetName() << "with pid"
                                  << requestor->GetPid();

            StateClientImpl::GetExecutionErrorRequest request;
            UnpackGetExecutionErrorRequest(packedRequest, request);
            ProcessGetExecutionErrorRequest(request, requestor);
            break;
        }

        case StateClientOperation::kRegisterForNotification: {
            GetLogger().LogInfo() << "Received a RegisterForNotification request from" << requestor->GetName()
                                  << "with pid" << requestor->GetPid();

            StateClientImpl::NotificationRegistrationRequest request;
            UnpackNotificationRegistrationRequest(packedRequest, request);
            if (!request.notificationSocketPath.empty()) {
                notifierSocketsPaths_.push_back(request.notificationSocketPath);
            } else {
                GetLogger().LogError() << "Received invalid RegisterForNotification request.";
            }

            break;
        }

        case StateClientOperation::kUnregisterFromNotification: {
            GetLogger().LogInfo() << "Received an UnregisterFromNotification request from" << requestor->GetName()
                                  << "with pid" << requestor->GetPid();

            StateClientImpl::NotificationRegistrationRequest request;
            UnpackNotificationRegistrationRequest(packedRequest, request);
            if (!request.notificationSocketPath.empty()) {
                const auto endIt = std::remove(
                    notifierSocketsPaths_.begin(), notifierSocketsPaths_.end(), request.notificationSocketPath);
                if (endIt != notifierSocketsPaths_.end()) {
                    notifierSocketsPaths_.erase(endIt, notifierSocketsPaths_.end());
                } else {
                    GetLogger().LogError() << "Invalid UnregisterFromNotification request, the socket path "
                                              "received is not registered.";
                }
            } else {
                GetLogger().LogError() << "Received invalid UnregisterFromNotification request.";
            }
            break;
        }

        case StateClientOperation::kGetInitialMachineStateTransitionRequest: {
            GetLogger().LogInfo() << "Received a GetInitialMachineStateTransition request from" << requestor->GetName()
                                  << "with pid" << requestor->GetPid();

            StateClientImpl::GetInitialMachineStateTransitionResponse response;
            if (was_initial_machine_state_transition_successful_) {
                response.status = StateClientImpl::GetInitialMachineStateTransitionResponse::Status::kSuccess;
            } else {
                response.status = StateClientImpl::GetInitialMachineStateTransitionResponse::Status::kFailed;
            }

            SendGetInitialMachineStateTransitionResponse(requestor, response);
            break;
        }

        default: {
            SendUnknownResponse(requestor);
            break;
        }
        }
    }
}
#endif
#if 0
bool ExecutionManager::ProcessSetStateRequest(ProcessedSetStateRequest& processedRequest)
{
    const StateClientImpl::SetStateRequest& request = processedRequest.request;
    std::shared_ptr<StartupConfiguration> requestor = processedRequest.requestor;

    const bool isMachineStateChange = request.change.functionGroupName == kMachineFunctionGroupName;

    const bool isMachineOffRequest = isMachineStateChange && config::kStateOff == request.change.newStateName;
    if (isMachineOffRequest) {
        /// @uptrace{SWS_EM_02549, 561714cde8d7828a7e2f14eb4d9f30ace84a8bfe}
        StateClientImpl::SetStateResponse response{StateClientImpl::SetStateResponse::Status::kInvalid, {}};
        GetLogger().LogError() << "MachineFG cannot be switched to Off state. Transition rejected.";

        SendSetStateResponse(requestor, response, request.responseTimeout);
        return true;
    }

    const bool isMachineVerifyRequest = isMachineStateChange && request.change.newStateName == "Verify";
    if (isMachineVerifyRequest) {
        GetLogger().LogWarn()
            << "Received request for MachineFG state Verify. This is currently not supported. Ignoring the request.";
        StateClientImpl::SetStateResponse response{
            StateClientImpl::SetStateResponse::Status::kIgnored, {StateClientImpl::StateChangeStatus::kFailed}};

        SendSetStateResponse(requestor, response, request.responseTimeout);
        return true;
    }

    const bool isShutdownRequest = isMachineStateChange && config::kStateMachineRestart == request.change.newStateName;
    const bool isRestartRequest = isMachineStateChange && config::kStateMachineShutdown == request.change.newStateName;
    if (isShutdownRequest || isRestartRequest) {

        GetLogger().LogInfo() << "Got Shutdown or Restart request" << requestor->GetName();

        // Switch all function group to state Off, report to the client and
        // switch MachineState to Shutdown/Restart state
        for (const auto& group : function_groups_) {
            if (group.second.GetName() != kMachineFunctionGroupName) {
                StateClientImpl::StateChange offRequest{group.second.GetName(), config::kStateOff};
                ProcessedSetStateRequest processedOffRequest{};
                processedOffRequest.request.change = offRequest;
                state_change_handler_.SetCompleteState(processedOffRequest);
            }
        }

        // Reply the client with status Accepted
        StateClientImpl::SetStateResponse response{
            StateClientImpl::SetStateResponse::Status::kAccepted, {StateClientImpl::StateChangeStatus::kSuccess}};

        SendSetStateResponse(requestor, response, request.responseTimeout);

        ProcessedSetStateRequest machineFGRequest
            = ProcessedSetStateRequest(kMachineFunctionGroupName, request.change.newStateName);

        // Set the machine state
        state_change_handler_.SetCompleteState(machineFGRequest);

        // Indicate that Run() should exit its loop
        platform_state_ = isShutdownRequest ? PlatformStates::kShuttingdown : PlatformStates::kRestarting;
        return true;
    }

    const bool isVerifyRequest = request.change.newStateName == "Verify";
    if (isVerifyRequest && !processedRequest.reparsingIsHandled) {
        GetLogger().LogInfo() << "Received Verify request for FunctionGroup" << request.change.functionGroupName;
        GetLogger().LogInfo() << "First, reparse the Execution Manifests.";

        netaos::nosal::core::Result<void> reparsingResult = ReparseManifests();
        processedRequest.reparsingIsHandled = true;
        if (!reparsingResult) {
            GetLogger().LogError()
                << "Reparsing of Execution Manifests was not successful. Ignoring the Verify request.";
            StateClientImpl::SetStateResponse response{
                StateClientImpl::SetStateResponse::Status::kFailed, {StateClientImpl::StateChangeStatus::kFailed}};

            SendSetStateResponse(requestor, response, request.responseTimeout);
            return true;
        }

        GetLogger().LogInfo() << "Reparsing successful. Proceeding with Verify request.";
    }

    netaos::nosal::core::Optional<StateClientImpl::SetStateResponse> completedSetState
        = state_change_handler_.SetState(processedRequest);
    if (!completedSetState) {
        return false;
    }

    SendSetStateResponse(requestor, *completedSetState, request.responseTimeout);
    return true;
}

void ExecutionManager::ProcessGetExecutionErrorRequest(const StateClientImpl::GetExecutionErrorRequest& request,
    std::shared_ptr<StartupConfiguration> startupConfiguration)
{
    const auto it = function_groups_.find(request.functionGroupName);

    if (it == function_groups_.cend()) {
        StateClientImpl::GetExecutionErrorResponse response{
            StateClientImpl::GetExecutionErrorResponse::Status::kFailed, {}};
        SendGetExecutionErrorResponse(startupConfiguration, response);
    } else {
        const FunctionGroup& functionGroup = it->second;
        if (!functionGroup.IsCurrentStateFailed()) {
            ProcessExecutionStateReports();
            UpdateFunctionGroupsStates();
        }
        netaos::nosal::core::Optional<ExecutionError> executionErrorOptional = functionGroup.GetExecutionError();
        if (executionErrorOptional) {
            StateClientImpl::GetExecutionErrorResponse response{
                StateClientImpl::GetExecutionErrorResponse::Status::kSuccess, *executionErrorOptional};
            SendGetExecutionErrorResponse(startupConfiguration, response);
        } else {
            StateClientImpl::GetExecutionErrorResponse response{
                StateClientImpl::GetExecutionErrorResponse::Status::kFailed, {}};
            SendGetExecutionErrorResponse(startupConfiguration, response);
        }
    }
}

void ExecutionManager::SendGetExecutionErrorResponse(std::shared_ptr<StartupConfiguration> startupConfiguration,
    const StateClientImpl::GetExecutionErrorResponse& response)
{
    std::string packedResponse;

    PackGetExecutionErrorResponse(response, packedResponse);

    PipeMessage message{
        {StateClientOperation::kGetExecutionErrorResponse, static_cast<std::uint32_t>(packedResponse.size())}, {}};

    message.data.reserve(packedResponse.size());
    message.data.insert(message.data.cbegin(), packedResponse.cbegin(), packedResponse.cend());
    startupConfiguration->SendMessageToStateClient(std::move(message));
}

void ExecutionManager::SendSetStateResponse(std::shared_ptr<StartupConfiguration> startupConfiguration,
    const StateClientImpl::SetStateResponse& response,
    std::chrono::time_point<std::chrono::steady_clock> responseTimeout)
{
    std::string packedResponse;
    PackSetStateResponse(response, packedResponse);

    PipeMessage message{
        {StateClientOperation::kSetStateResponse, static_cast<std::uint32_t>(packedResponse.size())}, {}};

    message.data.reserve(packedResponse.size());
    message.data.insert(message.data.cbegin(), packedResponse.cbegin(), packedResponse.cend());

    if (std::chrono::steady_clock::now() < responseTimeout) {
        startupConfiguration->SendMessageToStateClient(std::move(message));
    }
}

void ExecutionManager::SendGetInitialMachineStateTransitionResponse(std::shared_ptr<StartupConfiguration> requestor,
    const StateClientImpl::GetInitialMachineStateTransitionResponse& response)
{
    std::string packedData;
    PackGetInitialMachineStateTransitionResponse(response, packedData);

    PipeMessage message{{StateClientOperation::kGetInitialMachineStateTransitionResponse,
                            static_cast<std::uint32_t>(packedData.size())},
        {}};
    message.data.reserve(packedData.size());
    message.data.insert(message.data.begin(), packedData.begin(), packedData.end());
    requestor->SendMessageToStateClient(std::move(message));
}

void ExecutionManager::SendUnknownResponse(std::shared_ptr<StartupConfiguration> requestor)
{
    GetLogger().LogError() << "Received an invalid StateClient request!";

    PipeMessage message{{StateClientOperation::kUnknownResponse, 0}, {}};
    requestor->SendMessageToStateClient(std::move(message));
}

void ExecutionManager::InitializeFindProcess()
{
    GetLogger().LogInfo() << "Initialize FindProcessWrapper.";
    findProcess_ = std::make_unique<netaos::nosal::exec::internal::findprocess::EmFindProcessWrapper>();
    if (!findProcess_->Initialize(executablesHandler_->GetListOfExecutables())) {
        findProcess_.reset();
    }
}
#endif
void ExecutionManager::ProcessExecutionStateReports()
{
    if (!executed_startup_configurations_.empty()) {
#if 0
        StartupConfiguration::ProcessExecutionStateReports(executed_startup_configurations_);
#endif
    }
}

void ExecutionManager::NotifyUndefinedState(const FunctionGroup& functionGroup)
{
    const netaos::nosal::core::Optional<netaos::nosal::exec::ExecutionError> executionError = functionGroup.GetExecutionError();
    if (!executionError) {
        GetLogger().LogError() << "State of FunctionGroup" << functionGroup.GetName()
                               << "incorrectly recognized as Undefined Function Group State.";
        return;
    }

#if 0
    StateClientImpl::UpdateStateNotification notification{functionGroup.GetName(), *executionError};

    std::string packedNotification;
    PackUpdateStateNotification(notification, packedNotification);

    GetLogger().LogInfo() << "Notifying StateClient about Undefined State in Function Group" << functionGroup.GetName();
    const std::uint32_t messageTag = static_cast<std::uint32_t>(StateClientOperation::kUpdateStateNotification);
    const netaos::nosal::core::String message(std::to_string(messageTag) + packedNotification);

    std::for_each(
        notifierSocketsPaths_.cbegin(), notifierSocketsPaths_.cend(), [&message](const netaos::nosal::core::String& path) {
            LocalSocket localSocket(path);
            localSocket.Send(message);
        });
#endif
}


void ExecutionManager::UpdateFunctionGroupsStates()
{
    std::for_each(function_groups_.begin(), function_groups_.end(), [this](auto& group) {
        // dont update FunctionGroups that are in a transition controlled by EM.
        const bool is_not_in_transition = processed_requests_.end()
            == std::find_if(
                processed_requests_.begin(), processed_requests_.end(), [&group](ProcessedSetStateRequest request) {
                    return group.second.GetName() == request.functionGroup->GetName();
                });
        if (is_not_in_transition) {
            // check if an undefined state occured during a state transition or since the last check.
            const bool unhandledUndefinedState = group.second.HasUnhandledStateFailedEvent();
            const bool wasUndefinedState = group.second.IsCurrentStateFailed();
            group.second.UpdateState();
            if (unhandledUndefinedState || (!wasUndefinedState && group.second.IsCurrentStateFailed())) {
                this->NotifyUndefinedState(group.second);
            }
        }
    });
}

netaos::nosal::core::Result<void> ExecutionManager::LoadMachineConfiguration()
{
    GetLogger().LogInfo() << "Loading Machine Manifest.";

    manifest_.reset(new config::MachineManifest((root_path_ + kMachineManifestPath).c_str()));

    const std::vector<config::State>& machine_states = manifest_->GetMachineStates();
    if (machine_states.empty()) {
        GetLogger().LogFatal() << "Invalid definition of Machine States! Check Machine State configuration!";
        throw Exception("Invalid configuration");
    }

    function_groups_.emplace(kMachineFunctionGroupName, FunctionGroup(kMachineFunctionGroupName, machine_states));

    const std::vector<config::FunctionGroup>& groups = manifest_->GetFunctionGroups();
    for (const auto& group : groups) {
        function_groups_.emplace(group.GetName(), FunctionGroup(group.GetName(), group.GetStates()));
    }

    machine_specific_environment_variables_ = manifest_->GetEnvironmentVariables();

    GetLogger().LogInfo() << "Loading Machine Manifest finished.";
}

netaos::nosal::core::Result<void> ExecutionManager::LoadExecutionManifests()
{

    ExecutablesParser executables_parser(root_path_, ucmProcessesFilePath, manifest_->GetDefaultProcessTimeout());

    executables_handler_ = std::make_unique<ExecutablesHandler>(
        function_groups_, std::move(executables_parser), machine_specific_environment_variables_);

    netaos::nosal::core::Result<void> parsing_result = executables_handler_->ParseExecutionManifests();

    return parsing_result;
}

#if 0 
netaos::nosal::core::Result<void> ExecutionManager::ReparseManifests()
{
    if (executablesHandler_ == nullptr) {
        GetLogger().LogError() << "ExecutablesHandler not properly initialized.";
        return netaos::nosal::core::Result<void>{ExecManifestParsingErrc::kNotInitialized};
    }
    netaos::nosal::core::Result<void> reparsingResult = executablesHandler_->ReparseExecutionManifests();

    return reparsingResult;
}
void ExecutionManager::InitializeCgroups()
{
    GetLogger().LogInfo() << "Initialize Cgroups";

    const std::vector<config::ResourceGroup>& resourceGroups = manifest_->GetResourceGroups();
    if (resourceGroups.empty()) {
        GetLogger().LogError() << "Invalid definition of Resource Groups! Check Resource Group configuration!";
        return;
    }

    // to be able to create/access cgroups the file system has to be mounted first
    if (!PrepareFileSystem()) {
        GetLogger().LogError() << "Unable to mount file system!";
        return;
    }

    // Initialize libcgroup
    int ret = cgroup_init();
    if (ret) {
        GetLogger().LogError() << "libcgroup initialization failed with" << cgroup_strerror(ret);
        return;
    }

    for (const auto& resourceGroup : resourceGroups) {

        cgroup* new_cgroup = cgroup_new_cgroup(resourceGroup.GetName().c_str());
        if (!new_cgroup) {
            GetLogger().LogError() << "Cannot create new cgroup" << resourceGroup.GetName();
            continue;
        }

        // set cpu limits
        cgroup_controller* cgc_cpu = cgroup_add_controller(new_cgroup, "cpu");
        if (!cgc_cpu) {
            GetLogger().LogError() << "The cgroup" << resourceGroup.GetName() + ":"
                                   << "CPU controller cannot be added";
            continue;
        }

        // rt_runtime_us lower cpu limit
        // total rt_runtime_us = 1000000
        // ret = cgroup_add_value_uint64(cgc_cpu, "cpu.rt_runtime_us", resourceGroup.GetCpuUsage()*10000);
        // cfs_quota_us upper cpu limit
        // total cfs_quota_us = 100000
        ret = cgroup_add_value_uint64(cgc_cpu, "cpu.cfs_quota_us", resourceGroup.GetCpuUsage() * 1000);
        if (ret) {
            GetLogger().LogWarn() << "The cgroup" << resourceGroup.GetName() + ":"
                                  << "cannot add value to cpu.cfs_quota_us";
        }

        // set memory limits
        cgroup_controller* cgc_mem = cgroup_add_controller(new_cgroup, "memory");
        if (!cgc_mem) {
            GetLogger().LogError() << "The cgroup" << resourceGroup.GetName() + ":"
                                   << "memory controller cannot be added";
            continue;
        }

        ret = cgroup_add_value_int64(cgc_mem, "memory.limit_in_bytes", resourceGroup.GetMemUsage());
        if (ret) {
            GetLogger().LogWarn() << "The cgroup" << resourceGroup.GetName() + ":"
                                  << "cannot add value to memory.limit_in_bytes";
        }

        // create new cgroup
        ret = cgroup_create_cgroup(new_cgroup, 1);
        if (ret) {
            GetLogger().LogError() << "The cgroup" << resourceGroup.GetName() + ":"
                                   << "create cgroup failed with " << cgroup_strerror(ret);
            continue;
        }
        cgroup_free(&new_cgroup);
    }

    // special handling for Resource group EM
    // TODO: The configuration of the Execution Manager should be read out of an own configuration file - not
    // manifest
    cgroup* cgroupExec = cgroup_new_cgroup("Resource group EM");
    if (!cgroupExec) {
        GetLogger().LogError() << "Cannot create cgroup Resource group EM";
        return;
    }

    // if not exists create new one
    if (cgroup_get_cgroup(cgroupExec)) {
        GetLogger().LogWarn() << "Resource group Execution Manager doesn't exist. Create one with default values";

        // set cpu default limits
        cgroup_controller* cgcCpuExec = cgroup_add_controller(cgroupExec, "cpu");
        if (!cgcCpuExec) {
            GetLogger().LogError() << "A cgroup Resource group cpu controller cannot be added";
            return;
        }

        ret = cgroup_add_value_uint64(cgcCpuExec, "cpu.cfs_quota_us", 90 * 1000);
        if (ret) {
            GetLogger().LogWarn() << "A cgroup Resource group cannot add value to cpu.cfs_quota_us";
        }

        // set memory default limits
        cgroup_controller* cgcMemExec = cgroup_add_controller(cgroupExec, "memory");
        if (!cgcMemExec) {
            GetLogger().LogError() << "A cgroup Resource group memory controller cannot be added";
            return;
        }

        ret = cgroup_add_value_int64(cgcMemExec, "memory.limit_in_bytes", 500000000);
        if (ret) {
            GetLogger().LogWarn() << "A cgroup Resource group cannot add value to memory.limit_in_bytes";
        }

        ret = cgroup_create_cgroup(cgroupExec, 1);
        if (ret) {
            GetLogger().LogError() << "Creating of resource group Execution Manager failed with"
                                   << cgroup_strerror(ret);
            return;
        }
    }
    ret = cgroup_attach_task(cgroupExec);
    if (ret) {
        GetLogger().LogError() << "ExecutionManager: attach cgroup failed with" << cgroup_strerror(ret);
        return;
    }
    cgroup_free(&cgroupExec);

    GetLogger().LogInfo() << "Initialize Cgroups - done";
}

bool ExecutionManager::PrepareFileSystem()
{
    GetLogger().LogInfo() << "Preparing the FileSystem";

    // mount /proc
    if (-1 == mount("proc", "/proc", "proc", MS_MGC_VAL | MS_SILENT, "")) {
        GetLogger().LogError() << "/proc: mount failed!";
        return false;
    } else {
        GetLogger().LogDebug() << "/proc: successfully mounted";
    }

    // mount /sys
    if (-1 == mount("sysfs", "/sys", "sysfs", MS_MGC_VAL | MS_SILENT, "")) {
        GetLogger().LogError() << "/sys: mount failed!";
        return false;
    } else {
        GetLogger().LogDebug() << "/sys: successfully mounted";
    }

    // mount /sys/fs/cgroup
    if (-1 == mount("tmpfs", "/sys/fs/cgroup", "tmpfs", MS_MGC_VAL | MS_SILENT, "mode=755")) {
        GetLogger().LogError() << "/sys/fs/cgroup: mount failed!";
        return false;
    } else {
        GetLogger().LogDebug() << "/sys/fs/cgroup: successfully mounted";
    }

    // mount /sys/fs/cgroup/cpu
    std::string cpuDir = "/sys/fs/cgroup/cpu";
    mkdir(cpuDir.c_str(), S_IRUSR | S_IXUSR | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
    if (-1 == mount("cgroup", cpuDir.c_str(), "cgroup", MS_MGC_VAL | MS_SILENT, "cpu")) {
        GetLogger().LogError() << "/sys/fs/cgroup/cpu: mount failed!";
        return false;
    } else {
        GetLogger().LogDebug() << "/sys/fs/cgroup/cpu: successfully mounted";
    }

    // mount /sys/fs/cgroup/mem
    std::string memDir = "/sys/fs/cgroup/mem";
    mkdir(memDir.c_str(), S_IRUSR | S_IXUSR | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
    if (-1 == mount("cgroup", memDir.c_str(), "cgroup", MS_MGC_VAL | MS_SILENT, "memory")) {
        GetLogger().LogError() << "/sys/fs/cgroup/mem: mount failed!";
        return false;
    } else {
        GetLogger().LogDebug() << "/sys/fs/cgroup/mem: successfully mounted";
    }

    GetLogger().LogInfo() << "Finished preparing FileSystem.";
    return true;
}

void ExecutionManager::PrepareStateClientIpcDir()
{
    if (::mkdir(kStateClientIpcFilesPath, 0644) != 0) {
        if (errno != EEXIST) {
            GetLogger().LogError() << "Failed to create directory:" << kStateClientIpcFilesPath
                                   << "due to error:" << strerror(errno);
        } else {
            // the directory already exists, remove ipc files if any exists there
            DIR* directoryStream = ::opendir(kStateClientIpcFilesPath);

            while (true) {
                dirent* nextFile = ::readdir(directoryStream);

                if (nextFile == nullptr) {
                    break;
                }

                const netaos::nosal::core::String filePath(
                    (netaos::nosal::core::String(kStateClientIpcFilesPath) + "/") + nextFile->d_name);
                std::remove(filePath.c_str());
            }

            ::closedir(directoryStream);
        }
    }
}
#endif

}  // namespace internal
}  // namespace exec
}  // namespace nosal
}  // namespace netaos
